package com.fourth.utils;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.util.Base64;
import java.util.Properties;

public class MyKafkaUtil {
    private static Properties properties = new Properties();
    private static final String KAFKA_SERVER = "47.92.145.13:9092";

    static {
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,KAFKA_SERVER);
    }

    public static FlinkKafkaConsumer<String> getFlinkKafkaConsumer (String topic,String groupId){
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG,groupId);

        return  new FlinkKafkaConsumer<String>(
                topic,
                new KafkaDeserializationSchema<String>() {
                    @Override
                    public boolean isEndOfStream(String nextElement) {
                        return false;
                    }

                    @Override
                    public String deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
                        if (record == null || record.value() == null){
                            return null;
                        } else {
                            return new String(record.value());
                        }

                    }

                    @Override
                    public TypeInformation<String> getProducedType() {
                        return BasicTypeInfo.STRING_TYPE_INFO;
                    }
                },
                properties
        );
    }

    public static FlinkKafkaProducer<String> getFlinkKafkaProducer (String topic){

        return new FlinkKafkaProducer<String>(topic,new SimpleStringSchema(),properties);
    }

    public static String getInsertKafkaDDL (String topic){
        return "" +
                "WITH ( " +
                "  'connector' = 'kafka', " +
                "  'topic' = '" + topic + "', " +
                "  'properties.bootstrap.servers' = '"+KAFKA_SERVER+"', " +
                "  'format' = 'json' " +
                ")";
    }

    public static String getKafkaDDL(String topic,String groupId){
        return "WITH (" +
                " 'connector' = 'kafka', " +
                "  'topic' = '"+ topic +"', " +
                "  'properties.bootstrap.servers' = '"+ KAFKA_SERVER +"', " +
                "  'properties.group.id' = '"+ groupId +"', " +
                "  'scan.startup.mode' = 'earliest-offset', " +
                "  'format' = 'json' )";
    }

    public static String getUpsertKafka (String topic) {
        return "WITH (" +
                "  'connector' = 'upsert-kafka', " +
                "  'topic' = '"+ topic +"', " +
                "  'properties.bootstrap.servers' = '"+ KAFKA_SERVER +"', " +
                "  'key.format' = 'json', " +
                "  'value.format' = 'json' )";
    }

    public static String getKafkaSinkDDL(String topic) {
        return "WITH ( 'connector' = 'kafka', " +
                " 'topic' = '" + topic + "', " +
                " 'properties.bootstrap.servers' = '" + KAFKA_SERVER + "', " +
                " 'format' = 'json' " +
                ")";
    }
}
